"""ETF 数据获取模块 - 使用 AkShare."""

from __future__ import annotations

import logging
import math
import re
from datetime import datetime, timedelta
from typing import Any, Optional

import akshare as ak
import pandas as pd
import requests

from src.data.schemas import ETFHistoricalDataset, ETFMetadata, ETFRealtimeQuote
from src.utils import extract_exchange, normalize_symbol, now_shanghai


logger = logging.getLogger(__name__)


# AkShare 返回的必需列（支持中文和英文列名）
MIN_REQUIRED_COLUMNS_AKSHARE_CN = {"日期", "开盘", "收盘", "最高", "最低", "成交量"}
MIN_REQUIRED_COLUMNS_AKSHARE_EN = {"date", "open", "close", "high", "low", "volume"}


def _safe_float(value: Any) -> Optional[float]:
    """安全地将值转换为浮点数."""
    try:
        result = float(value)
    except (TypeError, ValueError):
        return None
    if math.isnan(result):
        return None
    return result


def _extract_code_suffix(symbol: str) -> Optional[str]:
    """提取 ETF 代码的后三位数字.

    Examples
    --------
    >>> _extract_code_suffix("159796")
    '796'
    >>> _extract_code_suffix("sz159796")
    '796'
    >>> _extract_code_suffix("159796.SZ")
    '796'
    >>> _extract_code_suffix("SPY")
    None
    """
    # 移除交易所后缀
    symbol = symbol.split(".")[0]
    # 移除前缀（如 sz, sh 等）
    match = re.search(r"(\d{6})$", symbol)
    if match:
        return match.group(1)[-3:]  # 返回后三位
    return None


def _fetch_sohu_realtime_price(symbol: str) -> Optional[dict[str, Any]]:
    """从搜狐财经获取实时价格数据.

    Parameters
    ----------
    symbol:
        ETF 基金代码，例如 `159796`、`sz159796` 等。

    Returns
    -------
    dict 或 None
        包含 price, as_of, volume 的字典，如果获取失败返回 None。
    """
    code_suffix = _extract_code_suffix(symbol)
    if not code_suffix:
        logger.debug("无法提取代码后缀，跳过搜狐财经接口：symbol=%s", symbol)
        return None

    # 提取原始代码（移除交易所后缀和前缀）
    raw_code = symbol.split(".")[0]
    # 移除前缀（如 sz, sh 等）
    match = re.search(r"(\d{6})$", raw_code)
    if not match:
        logger.debug("无法提取6位数字代码，跳过搜狐财经接口：symbol=%s", symbol)
        return None
    raw_code = match.group(1)

    url = f"https://hq.stock.sohu.com/cn/{code_suffix}/cn_{raw_code}-4.html"
    logger.debug("正在从搜狐财经获取实时价格：symbol=%s, url=%s", symbol, url)

    try:
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        }
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        content = response.text

        # 查找 time_data 函数调用
        pattern = r"time_data\(\[(.*?)\]\)"
        match = re.search(pattern, content, re.DOTALL)
        if not match:
            logger.warning("未找到 time_data 数据：symbol=%s", symbol)
            return None

        # 解析 JavaScript 数组数据
        data_str = match.group(1)
        # 使用正则表达式提取所有数组
        array_pattern = r"\[(.*?)\]"
        arrays = re.findall(array_pattern, data_str)

        if not arrays or len(arrays) < 2:
            logger.warning("数据格式异常：symbol=%s, arrays_count=%s", symbol, len(arrays))
            return None

        # 第一个数组是汇总信息，跳过
        # 从第二个数组开始是分钟级数据
        # 格式：['时间', '价格1', '价格2', '成交量', '其他']
        latest_data = arrays[-1]  # 取最后一个数组（最新数据）
        parts = [p.strip().strip("'\"") for p in latest_data.split(",")]

        if len(parts) < 3:
            logger.warning("数据字段不足：symbol=%s, parts=%s", symbol, parts)
            return None

        time_str = parts[0]  # 时间，格式如 '14:53'
        price_str = parts[1] if len(parts) > 1 else parts[0]  # 价格
        volume_str = parts[3] if len(parts) > 3 else None  # 成交量

        # 解析价格
        try:
            price = float(price_str)
        except (ValueError, TypeError):
            logger.warning("价格解析失败：symbol=%s, price_str=%s", symbol, price_str)
            return None

        # 解析成交量
        volume: Optional[float] = None
        if volume_str:
            try:
                volume = float(volume_str)
            except (ValueError, TypeError):
                pass

        # 构建时间戳（结合当前日期）
        now = now_shanghai()
        try:
            hour, minute = map(int, time_str.split(":"))
            as_of = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
            # 如果时间晚于当前时间，说明是昨天的数据
            if as_of > now:
                as_of = as_of - timedelta(days=1)
        except (ValueError, AttributeError) as e:
            logger.warning("时间解析失败：symbol=%s, time_str=%s, error=%s", symbol, time_str, e)
            as_of = now

        logger.info(
            "搜狐财经数据获取成功：symbol=%s, price=%s, time=%s, volume=%s",
            symbol,
            price,
            as_of.strftime("%H:%M:%S"),
            volume,
        )

        return {
            "price": price,
            "as_of": as_of,
            "volume": volume,
        }

    except requests.RequestException as e:
        logger.warning("搜狐财经请求失败：symbol=%s, error=%s", symbol, e)
        return None
    except Exception as e:  # noqa: BLE001
        logger.warning("搜狐财经数据解析失败：symbol=%s, error=%s", symbol, e)
        return None


def _sanitize_akshare_frame(frame: pd.DataFrame) -> pd.DataFrame:
    """清洗 AkShare 返回的历史数据.

    Parameters
    ----------
    frame:
        AkShare 返回的 DataFrame，可能包含中文列名或英文列名

    Returns
    -------
    清洗后的 DataFrame，统一转换为中文列名格式
    """
    if frame.empty:
        raise ValueError("未获取到任何历史数据")

    cleaned = frame.copy()
    
    # 检测列名格式（中文或英文）
    has_cn_columns = "日期" in cleaned.columns or "开盘" in cleaned.columns
    has_en_columns = "date" in cleaned.columns or "open" in cleaned.columns
    
    # 检查必需列
    if has_cn_columns:
        missing_columns = MIN_REQUIRED_COLUMNS_AKSHARE_CN.difference(cleaned.columns)
        if missing_columns:
            raise ValueError(f"历史数据缺少必要字段: {missing_columns}")
    elif has_en_columns:
        missing_columns = MIN_REQUIRED_COLUMNS_AKSHARE_EN.difference(cleaned.columns)
        if missing_columns:
            raise ValueError(f"历史数据缺少必要字段: {missing_columns}")
        # 将英文列名转换为中文列名
        column_mapping = {
            "date": "日期",
            "open": "开盘",
            "high": "最高",
            "low": "最低",
            "close": "收盘",
            "volume": "成交量",
        }
        cleaned = cleaned.rename(columns=column_mapping)
    else:
        raise ValueError("无法识别数据格式：既没有中文列名也没有英文列名")
    
    # 确保日期列存在并设置为索引
    if "日期" in cleaned.columns:
        # 将日期列转换为 datetime 并设置为索引
        cleaned["日期"] = pd.to_datetime(cleaned["日期"])
        cleaned = cleaned.set_index("日期")
    elif cleaned.index.name == "日期" or isinstance(cleaned.index, pd.DatetimeIndex):
        # 索引已经是日期
        cleaned.index = pd.to_datetime(cleaned.index)
    else:
        raise ValueError("无法找到日期列或日期索引")

    # 排序并去除全为 NaN 的行
    cleaned = cleaned.sort_index()
    cleaned = cleaned.dropna(how="all")
    
    # 确保数据类型正确
    numeric_columns = ["开盘", "收盘", "最高", "最低", "成交量"]
    for col in numeric_columns:
        if col in cleaned.columns:
            cleaned[col] = pd.to_numeric(cleaned[col], errors="coerce")
    
    # 如果有成交额、振幅等列，也转换为数值类型
    optional_numeric_columns = ["成交额", "振幅", "涨跌幅", "涨跌额", "换手率"]
    for col in optional_numeric_columns:
        if col in cleaned.columns:
            cleaned[col] = pd.to_numeric(cleaned[col], errors="coerce")

    return cleaned


def _fetch_metadata_akshare(symbol: str) -> ETFMetadata:
    """使用 AkShare 获取 ETF 基础信息.

    Parameters
    ----------
    symbol:
        ETF 代码（6位数字）

    Returns
    -------
    ETFMetadata
        ETF 元数据
    """
    name: Optional[str] = None
    exchange: Optional[str] = None
    nav_price: Optional[float] = None
    previous_close: Optional[float] = None

    try:
        # 尝试从 fund_etf_fund_daily_em 获取当前交易日所有 ETF 信息
        daily_funds = ak.fund_etf_fund_daily_em()
        if not daily_funds.empty and "基金代码" in daily_funds.columns:
            # 查找匹配的 ETF
            fund_info = daily_funds[daily_funds["基金代码"] == symbol]
            if not fund_info.empty:
                row = fund_info.iloc[0]
                name = row.get("基金简称") if "基金简称" in row else None
                nav_price = _safe_float(row.get("最新净值")) if "最新净值" in row else None
                previous_close = _safe_float(row.get("最新净值")) if "最新净值" in row else None
                logger.debug("从 fund_etf_fund_daily_em 获取到 ETF 信息: %s", symbol)
    except Exception as e:
        logger.debug("从 fund_etf_fund_daily_em 获取信息失败: %s", e)

    # 如果未获取到名称，尝试从实时行情获取
    if not name:
        try:
            spot_data = ak.fund_etf_spot_em()
            if not spot_data.empty and "代码" in spot_data.columns:
                fund_info = spot_data[spot_data["代码"] == symbol]
                if not fund_info.empty:
                    row = fund_info.iloc[0]
                    name = row.get("名称") if "名称" in row else None
                    previous_close = _safe_float(row.get("最新价")) if "最新价" in row else None
                    logger.debug("从 fund_etf_spot_em 获取到 ETF 信息: %s", symbol)
        except Exception as e:
            logger.debug("从 fund_etf_spot_em 获取信息失败: %s", e)

    # 推断交易所
    if symbol.startswith(("50", "51", "58")):
        exchange = "SSE"  # 上海证券交易所
    elif symbol.startswith(("15", "16", "18")):
        exchange = "SZSE"  # 深圳证券交易所
    else:
        exchange = extract_exchange(symbol)

    return ETFMetadata(
        symbol=symbol,
        name=name,
        exchange=exchange,
        fund_family=None,
        category=None,
        currency="CNY",
        total_assets=None,
        shares_outstanding=None,
        nav_price=nav_price,
        previous_close=previous_close,
        source="akshare",
    )


def _fetch_etf_history_akshare(
    symbol: str,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
) -> pd.DataFrame:
    """使用 AkShare 获取 ETF 历史数据.

    Parameters
    ----------
    symbol:
        ETF 代码（6位数字）
    start_date:
        开始日期（YYYY-MM-DD 格式）
    end_date:
        结束日期（YYYY-MM-DD 格式）

    Returns
    -------
    DataFrame
        AkShare 格式的历史数据（中文列名）
    """
    # 转换日期格式：YYYY-MM-DD -> YYYYMMDD
    start_str = None
    end_str = None
    if start_date:
        try:
            dt = datetime.strptime(start_date, "%Y-%m-%d")
            start_str = dt.strftime("%Y%m%d")
        except ValueError:
            logger.warning("开始日期格式错误，使用默认值: %s", start_date)
    
    if end_date:
        try:
            dt = datetime.strptime(end_date, "%Y-%m-%d")
            end_str = dt.strftime("%Y%m%d")
        except ValueError:
            logger.warning("结束日期格式错误，使用默认值: %s", end_date)

    # 优先使用 fund_etf_hist_sina（获取所有历史数据，数据更完整）
    # 需要添加交易所前缀
    if symbol.startswith(("50", "51", "58")):
        sina_symbol = f"sh{symbol}"
    elif symbol.startswith(("15", "16", "18")):
        sina_symbol = f"sz{symbol}"
    else:
        sina_symbol = symbol
    
    df = None
    
    # 首先尝试使用 fund_etf_hist_sina
    try:
        df = ak.fund_etf_hist_sina(symbol=sina_symbol)
        if not df.empty:
            logger.debug("使用 fund_etf_hist_sina 获取数据成功: %s", symbol)
            # 如果指定了日期范围，进行过滤
            if start_str or end_str:
                if "date" in df.columns:
                    df["date"] = pd.to_datetime(df["date"])
                    if start_str:
                        start_dt = datetime.strptime(start_str, "%Y%m%d")
                        df = df[df["date"] >= start_dt]
                    if end_str:
                        end_dt = datetime.strptime(end_str, "%Y%m%d")
                        df = df[df["date"] <= end_dt]
    except Exception as e:
        logger.debug("fund_etf_hist_sina 失败，尝试 fund_etf_hist_em: %s", e)
    
    # 如果 sina 失败或返回空数据，使用 fund_etf_hist_em
    if df is None or df.empty:
        try:
            df = ak.fund_etf_hist_em(
                symbol=symbol,
                period="daily",
                start_date=start_str or "19700101",
                end_date=end_str or "20500101",
                adjust="",
            )
            if not df.empty:
                logger.debug("使用 fund_etf_hist_em 获取数据成功: %s", symbol)
        except Exception as e:
            logger.error("AkShare 获取历史数据失败: symbol=%s, error=%s", symbol, e)
            raise
    
    if df.empty:
        raise ValueError(f"AkShare 返回空数据: {symbol}")
    
    return df


def fetch_etf_history(
    symbol: str,
    *,
    limit: Optional[int] = None,
    start: Optional[str] = None,
    end: Optional[str] = None,
) -> ETFHistoricalDataset:
    """获取 ETF 历史数据（使用 AkShare）.

    Parameters
    ----------
    symbol:
        ETF 基金代码，例如 `159919`、`sz159919` 或 `515790`。
    limit:
        返回的最大行数，如果为 None 则获取全量数据（默认 None）。
        如果指定，则只返回最近 limit 条数据。
    start, end:
        可选，指定日期范围（YYYY-MM-DD）。未指定则从最早可获取日期开始至当前。
    """
    if limit is not None and limit <= 0:
        raise ValueError("limit 必须为正整数或 None")

    normalized = normalize_symbol(symbol)
    logger.info("正在获取 ETF 历史数据（AkShare）：symbol=%s, normalized=%s", symbol, normalized)

    # 使用 AkShare 获取历史数据
    try:
        df = _fetch_etf_history_akshare(normalized, start_date=start, end_date=end)
        frame = _sanitize_akshare_frame(df)
    except Exception as e:
        logger.error("获取 ETF 历史数据失败: symbol=%s, error=%s", normalized, e)
        raise

    # 应用 limit 限制（如果指定）
    if limit is not None and len(frame) > limit:
        frame = frame.iloc[-limit:]
        logger.info("应用 limit 限制: 返回最近 %d 条数据", limit)

    # 获取元数据
    try:
        metadata = _fetch_metadata_akshare(normalized)
    except Exception as e:
        logger.warning("获取 ETF 元数据失败: %s", e)
        metadata = ETFMetadata(symbol=normalized, source="akshare")

    start_dt: datetime = frame.index.min().to_pydatetime()
    end_dt: datetime = frame.index.max().to_pydatetime()

    logger.info(
        "ETF 历史数据获取完成: symbol=%s, rows=%s, start=%s, end=%s",
        normalized,
        len(frame),
        start_dt.strftime("%Y-%m-%d"),
        end_dt.strftime("%Y-%m-%d"),
    )

    dataset = ETFHistoricalDataset(
        symbol=normalized,
        frame=frame,
        start=start_dt,
        end=end_dt,
        count=len(frame),
        metadata=metadata,
        source="akshare",
    )

    if dataset.count < 60:
        logger.warning("历史数据不足 60 条，可能影响波动率等指标计算")

    return dataset


def fetch_etf_intraday_price(symbol: str) -> ETFRealtimeQuote:
    """获取 ETF 盘中实时价格（使用 AkShare）.

    该函数优先尝试从搜狐财经获取实时价格和时间戳；如果失败，
    则尝试通过 AkShare 的实时行情接口获取。

    Parameters
    ----------
    symbol:
        ETF 基金代码

    Returns
    -------
    ETFRealtimeQuote
        实时行情数据
    """
    normalized = normalize_symbol(symbol)
    logger.info(
        "正在获取 ETF 盘中实时价格（AkShare）：symbol=%s, normalized=%s", symbol, normalized
    )

    # 优先尝试从搜狐财经获取
    sohu_data = _fetch_sohu_realtime_price(symbol)
    price: Optional[float] = None
    as_of = now_shanghai()
    volume: Optional[float] = None
    previous_close: Optional[float] = None
    change: Optional[float] = None
    change_percent: Optional[float] = None

    if sohu_data:
        price = sohu_data.get("price")
        as_of = sohu_data.get("as_of", as_of)
        volume = sohu_data.get("volume")

    # 如果搜狐财经获取失败，使用 AkShare 实时行情
    if price is None:
        try:
            spot_data = ak.fund_etf_spot_em()
            if not spot_data.empty and "代码" in spot_data.columns:
                fund_info = spot_data[spot_data["代码"] == normalized]
                if not fund_info.empty:
                    row = fund_info.iloc[0]
                    price = _safe_float(row.get("最新价")) if "最新价" in row else None
                    volume = _safe_float(row.get("成交量")) if "成交量" in row else volume
                    previous_close = _safe_float(row.get("昨收")) if "昨收" in row else None
                    change_pct = _safe_float(row.get("涨跌幅")) if "涨跌幅" in row else None
                    
                    if price and previous_close:
                        change = price - previous_close
                        if change_pct is not None:
                            change_percent = change_pct / 100.0
                        elif previous_close != 0:
                            change_percent = change / previous_close
                    
                    logger.info("从 AkShare 实时行情获取到价格: symbol=%s, price=%s", normalized, price)
        except Exception as e:
            logger.warning("AkShare 实时行情获取失败: %s", e)

    if price is None:
        raise ValueError(f"无法获取 {symbol} 的盘中实时价格")

    return ETFRealtimeQuote(
        symbol=normalized,
        price=price,
        as_of=as_of,
        currency="CNY",
        volume=volume,
        previous_close=previous_close,
        change=change,
        change_percent=change_percent,
        source="akshare",
    )


def main():
    """测试函数."""
    quote = fetch_etf_intraday_price("159796")
    print("最新价格：", quote)
    history = fetch_etf_history("159796", limit=10)
    print("历史数据：", history.frame.head())


if __name__ == "__main__":
    main()
